package com.ruoyi.socket.gateway;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Throwables;
import com.ruoyi.socket.probuffer.DataInfo;
import com.ruoyi.socket.util.*;
import com.ruoyi.socket.util.anbaobr.SdkParse;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Handles a server-side channel.
 */
public class GateWayInboundHandler
        extends ChannelInboundHandlerAdapter {

    private static Logger logger = LoggerFactory.getLogger(GateWayInboundHandler.class);
    // (1) SimpleChannelInboundHandler类会默认自动释放资源
    //保留所有与服务器建立连接的channel对象
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    //    private Timer timer = new Timer();
    ScheduledExecutorService executor =
            Executors.newScheduledThreadPool(10);
    private RedisPool redisPool = RedisPool.getInstance();

    /**
     * chanelRead() 事件处理方法。每当从客户端收到新的数据时，这个方法会在收到消息时被调用
     * 收到的消息的类型是 ByteBuf
     *
     * @param ctx
     * @param msg
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        ByteBuf byteBuf = (ByteBuf) msg;
        String buf = ByteBufStringUtils.convertByteBufToString(byteBuf);
        String gateWayDate = buf.replace("\r\n", "");


        // todo 处理业务逻辑
        // 1 解析医疗网关处理的数据，校验网关的设备id以及设备类型是否符合约定，不符合直接黑名单
        // 2 通过医疗网关的content内容流，根据对应的br或者ecg协议解析数据，并存入redis队列
        // 3 若是客户端连接，（app，web等），校验该连接，取出redis解析好的数据。
        logger.info("接收远程发送的原始数据 {}", gateWayDate);
        FileUtils.writeByte(gateWayDate, "proto.txt");


        try {
            // 解析probuf
            byte[] bytes = ByteStringUtils.toByteArray(gateWayDate);
            DataInfo.DeviceData deviceData = DeviceDataInfoUtils.decode(bytes);
            String devicetype = deviceData.getDevicetype();
            String content = deviceData.getContent().toStringUtf8();
            logger.info("接收远程发送解析后的数据 {}", content);

            // 心电仪类型
            if ("1".equals(devicetype)) {
                FileUtils.writeByte(content, "ecg.data");


                // 呼吸机
            } else if ("0".equals(devicetype)) {
                // 解析呼吸机数据，并将解析好的数据放入redis
                FileUtils.writeByte(content, "br.data");
                // 呼吸机解析功能
                Map map = SdkParse.parseText(content);
                String result = JSON.toJSONString(map);
                redisPool.getConn().lpush("br", result);
                System.out.println("处理呼吸机并放入队列br  " + result);
            }
            ctx.writeAndFlush(byteBuf);
        } catch (Exception e) {


        }

        // 根据 传递的内容 判断是否是客户端连接 todo 校验前端数据的接口
        if ("br".equals(gateWayDate)) {


            executor.scheduleAtFixedRate(
                    new TimerTask() {
                        @Override
                        public void run() {
                            String br = redisPool.getConn().lpop("br");
                            // 若没有取出数据则不输出
                            if (null != br && !"".equals(br)) {
                                ByteBuf result = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer(br, Charset.forName("UTF-8")));
                                ctx.writeAndFlush(result);
                            }

                        }
                    }, 1, 1000, TimeUnit.MICROSECONDS
            );

        }

    }


    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        channelGroup.add(ctx.channel());
        ByteBuf buffer = ctx.channel().alloc().buffer();
        // 获取 buffer中的内容
        String sendMsg = ByteBufStringUtils.convertByteBufToString(buffer);
        System.out.println("通道激活时，接收远程发送数据" + sendMsg);

        ctx.fireChannelActive();

    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
        cause.printStackTrace();
        System.out.println("exceptionCaught... " + Throwables.getStackTraceAsString(cause));
        ctx.close();
    }




/*    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
//        ByteBufAllocator alloc = ctx.channel().alloc();
//        System.out.println("通道读取结束 channelReadComplete... " + ByteBufStringUtils.convertByteBufToString(alloc.buffer()));
    }*/


}